package com.xiaoyilin.serializer;

import com.xiaoyilin.domain.Company;
import org.apache.kafka.common.serialization.Deserializer;

import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.util.Map;

/**
 *
 */
public class CompanyDeserializer implements Deserializer<Company> {

    @Override
    public void configure(Map configs, boolean isKey) {

    }

    @Override
    public Company deserialize(String topic, byte[] data) {
        ByteBuffer buffer = ByteBuffer.wrap(data);
        int nameLen, addressLen;
        String name, address;
        nameLen = buffer.getInt();
        byte[] nameBytes = new byte[nameLen];
        buffer.get(nameBytes);
        addressLen = buffer.getInt();
        byte[] addressBytes = new byte[addressLen];
        buffer.get(addressBytes);
        try {
            name = new String(nameBytes,"UTF-8");
            address = new String(addressBytes,"UTF-8");
            return new Company(name,address);
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }
       return null;
    }

    @Override
    public void close() {

    }


}
